package com.cskaoyan.mall.search.service.impl;

import co.elastic.clients.elasticsearch._types.aggregations.AggregateBuilders;
import com.cskaoyan.mall.common.constant.RedisConst;
import com.cskaoyan.mall.product.dto.CategoryHierarchyDTO;
import com.cskaoyan.mall.product.dto.PlatformAttributeInfoDTO;
import com.cskaoyan.mall.product.dto.SkuInfoDTO;
import com.cskaoyan.mall.product.dto.TrademarkDTO;
import com.cskaoyan.mall.search.client.ProductApiClient;
import com.cskaoyan.mall.search.converter.GoodsConverter;
import com.cskaoyan.mall.search.dto.GoodsDTO;
import com.cskaoyan.mall.search.dto.SearchResponseAttrDTO;
import com.cskaoyan.mall.search.dto.SearchResponseDTO;
import com.cskaoyan.mall.search.dto.SearchResponseTmDTO;
import com.cskaoyan.mall.search.model.Goods;
import com.cskaoyan.mall.search.model.SearchAttr;
import com.cskaoyan.mall.search.param.SearchParam;
import com.cskaoyan.mall.search.repository.GoodsRepository;
import com.cskaoyan.mall.search.service.SearchService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.lucene.search.join.ScoreMode;
import org.elasticsearch.index.query.*;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.nested.Nested;
import org.elasticsearch.search.aggregations.bucket.nested.NestedAggregationBuilder;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.redisson.api.RScoredSortedSet;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.query.FetchSourceFilter;
import org.springframework.data.elasticsearch.core.query.HighlightQueryBuilder;
import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

/**
 * @Author Pudding
 * @Date 2024/7/12 21:43
 */
@Slf4j
@Service
public class SearchServiceImpl implements SearchService {

    @Autowired
    private ProductApiClient productApiClient;
    @Autowired
    private GoodsRepository goodsRepository;
    @Autowired
    private ElasticsearchRestTemplate elasticsearchRestTemplate;
    @Autowired
    private GoodsConverter goodsConverter;
    @Autowired
    RedissonClient redissonClient;

    /**
     * 上架商品
     * 1. 根据商品Id查询出商品信息
     * 2. 存入ES中
     */
    @Override
    public void upperGoods(Long skuId) {

        //1. 根据商品id查询出商品信息，并封装到goods对象中
        Goods goods = new Goods();

        // 可以使用异步编排工具 CompletableFuture 给以下的4部操作进行异步化，提高效率
        // 1. 零依赖：supplyAsync()
        // 2. 一元依赖：thenAcceptAsync
        // 3. 一元依赖：thenAcceptAsync
        // 4. 零依赖: runAsync()
        // allOf(1,2,3,4);

        long startTime = System.currentTimeMillis();
        
        //创建线程池对象
        ExecutorService pool = Executors.newFixedThreadPool(5);

        CompletableFuture<SkuInfoDTO> cf1 = CompletableFuture.supplyAsync(() -> {
            //1.1 根据skuId查询出商品的基本信息（通过openFeign远程调用）
            SkuInfoDTO skuInfoDTO = productApiClient.getSkuInfo(skuId);
            goods.setId(skuInfoDTO.getId());
            goods.setTitle(skuInfoDTO.getSkuName());
            goods.setPrice(skuInfoDTO.getPrice().doubleValue());
            goods.setDefaultImg(skuInfoDTO.getSkuDefaultImg());

            return skuInfoDTO;
        }, pool);

        CompletableFuture<Void> cf2 = cf1.thenAcceptAsync(skuInfoDTO -> {
            //1.2 根据skuInfoDTO中品牌id查询出品牌信息（通过openFeign远程调用）
            TrademarkDTO trademarkDTO = productApiClient.getTrademark(skuInfoDTO.getTmId());
            goods.setTmId(trademarkDTO.getId());
            goods.setTmName(trademarkDTO.getTmName());
            goods.setTmLogoUrl(trademarkDTO.getLogoUrl());
        }, pool);

        CompletableFuture<Void> cf3 = cf1.thenAcceptAsync(skuInfoDTO -> {
            //1.3 根据skuInfoDTO中三级分类id查询出三级分类信息（通过openFeign远程调用）
            CategoryHierarchyDTO categoryView = productApiClient.getCategoryView(skuInfoDTO.getThirdLevelCategoryId());
            goods.setFirstLevelCategoryId(categoryView.getFirstLevelCategoryId());
            goods.setFirstLevelCategoryName(categoryView.getFirstLevelCategoryName());
            goods.setSecondLevelCategoryId(categoryView.getSecondLevelCategoryId());
            goods.setSecondLevelCategoryName(categoryView.getSecondLevelCategoryName());
            goods.setThirdLevelCategoryId(categoryView.getThirdLevelCategoryId());
            goods.setThirdLevelCategoryName(categoryView.getThirdLevelCategoryName());
        }, pool);

        CompletableFuture<Void> cf4 = CompletableFuture.runAsync(() -> {
            //1.4 根据skuId查询出平台属性的List
            List<PlatformAttributeInfoDTO> attrList = productApiClient.getAttrList(skuId);
            //通过流将List<PlatformAttributeInfoDTO>转化为goods中的List<SearchAttr>
            List<SearchAttr> searchAttrList = attrList.stream().map(PlatformAttributeInfoDTO -> {
                SearchAttr searchAttr = new SearchAttr();
                searchAttr.setAttrId(PlatformAttributeInfoDTO.getId());
                searchAttr.setAttrName(PlatformAttributeInfoDTO.getAttrName());
                searchAttr.setAttrValue(PlatformAttributeInfoDTO.getAttrValueList().get(0).getValueName());

                return searchAttr;
            }).collect(Collectors.toList());
            goods.setAttrs(searchAttrList);
        }, pool);

        CompletableFuture.allOf(cf2, cf3, cf4).join(); //join在哪个线程中执行，哪个线程等待

        //2. 把商品存入ES文档中
        goodsRepository.save(goods);

        long endTime = System.currentTimeMillis();
        log.info("商品上架总时间：{}毫秒，开始时间：{}，结束时间：{}，skuId：{}", (endTime - startTime), startTime, endTime, skuId);

    }

    /**
     * 下架商品
     * 1. 把对应的商品，从ES中删除
     */
    @Override
    public void lowerGoods(Long skuId) {
        //1. 删除文档：把对应的商品，从ES中删除
        goodsRepository.deleteById(skuId);
    }


    /*
    	1. 获取商品数据
    	2. 给热度值 + 1
    	3. 把最新的热度值 存入ES中
    */
    @Override
    public void incrHotScore(Long skuId) {
        /*********方式一：传统方法***********/
        /*
        //1. 获取商品数据对象
        Optional<Goods> optionalGoods = goodsRepository.findById(skuId);
        Goods goods = optionalGoods.get();
        //2. 给热度值+1
        goods.setHotScore(goods.getHotScore() + 1);
        //3. 把最新的热度值 存入ES中
        goodsRepository.save(goods);
        */


        /**
         * 1. 背景说明：
         *  当前接口是在商品服务中，每次访问商品详情页的时候，发起远程调用，调用搜索服务中更新热度的接口
         *  假如商品详情页访问量很大， 那么就意味着每一次访问都需要更新ES，这样就会对ES产生比较大的压力
         *  而热度这个数据，其实是用户不敏感的，没有必要实时更新
         *
         * 2. 改进的方案
         *   a) 把热度值使用Redis来存储
         *   b) 在用户访问商品详情页的时候，更新Redis中的热度值
         *   c) 当Redis中的热度值更新到一定程度的时候(比如说是10的倍数的时候)，再去把最新的热度值设置到ES中去
         *   d) 这样做的目的就是为了降低ES的更新频率(因为本质上来说，ES这个搜索引擎存储数据还是基于磁盘来存储的)，减轻ES的压力
         *
         * 3. 既然要把商品数据的热度值存入Redis，那么使用Redis的什么数据结构来存呢？
         *      string  set  list  hash  zset
         */
        /*********方式二：改进方法***********/
        //1. 获取热度值集合
        RScoredSortedSet<Object> sortedSet = redissonClient.getScoredSortedSet(RedisConst.HOT_SCORE);

        //2. 给热度值加1
        sortedSet.addScore(skuId, 1);

        //3. 判断当前的得分是不是10的倍数，是的话就写入ES中
        Double score = sortedSet.getScore(skuId);
        if (score.intValue() % 10 == 0){
            //把最新的得分写入ES中
            //4. 获取商品数据对象
            Optional<Goods> optionalGoods = goodsRepository.findById(skuId);
            Goods goods = optionalGoods.get();
            //5. 把最新的热度值 存入ES中
            goods.setHotScore(score.longValue());
            goodsRepository.save(goods);
        }


    }

    /**
     * 条件查询
     * 这个接口的实现，可以选择两种方式
     * 1. RestHighLevelClient
     * 2. ElasticSearchRestTemplate
     */
    @Override
    public SearchResponseDTO search(SearchParam searchParam) throws IOException {

        //3. 为搜索请求体对象构建搜索条件，调用下面的方法
        NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
        buildDSL(nativeSearchQueryBuilder, searchParam);

        //2. 构建搜索请求对象
        NativeSearchQuery nativeSearchQuery = nativeSearchQueryBuilder.build();

        //1. 发起HTTP搜索请求，获取HTTP搜索响应
        SearchHits<Goods> searchHits = elasticsearchRestTemplate.search(nativeSearchQuery, Goods.class);

        //4. 解析搜索的结果，调用下面的方法
        SearchResponseDTO searchResponseDTO = parseResponse(searchHits, searchParam);

        return searchResponseDTO;
    }

    /**
     * 为搜索请求体对象构建搜索条件
     *
     * @param nativeSearchQueryBuilder 搜索查询请求体对象
     * @param searchParam              前端传入的查询参数
     */
    private void buildDSL(NativeSearchQueryBuilder nativeSearchQueryBuilder, SearchParam searchParam) {

        //1. 构建bool_Query查询
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

        //1.1 关键字查询
        //1.1.1 先判断前端是否传来了关键字
        if (StringUtils.isNotBlank(searchParam.getKeyword())) {
            MatchQueryBuilder keywordQueryBuilder = QueryBuilders.matchQuery("title", searchParam.getKeyword());
            //1.1.2 将关键字查询添加到bool_Query的must中
            boolQueryBuilder.must(keywordQueryBuilder);
        }

        //1.2 品牌查询
        //1.2.1 先判断前端是否传入了品牌String
        if (StringUtils.isNotBlank(searchParam.getTrademark())) {
            String[] trademark = searchParam.getTrademark().split(":");
            TermQueryBuilder tmIdQueryBuilder = QueryBuilders.termQuery("tmId", Long.valueOf(trademark[0]));
            //1.2.2 将关键字查询添加到bool_Query的filter中
            boolQueryBuilder.filter(tmIdQueryBuilder);
        }

        //1.3 分类查询
        //1.3.1 先判断前端是否传入了一级/二级/三级分类id
        if (searchParam.getFirstLevelCategoryId() != null) {
            TermQueryBuilder firstLevelCategoryIdQueryBuilder = QueryBuilders.termQuery("firstLevelCategoryId", searchParam.getFirstLevelCategoryId());
            //1.3.2 将关键字查询添加到bool_Query的filter中
            boolQueryBuilder.filter(firstLevelCategoryIdQueryBuilder);
        }
        if (searchParam.getSecondLevelCategoryId() != null) {
            TermQueryBuilder secondLevelCategoryIdQueryBuilder = QueryBuilders.termQuery("secondLevelCategoryId", searchParam.getSecondLevelCategoryId());
            //1.3.3 将关键字查询添加到bool_Query的filter中
            boolQueryBuilder.filter(secondLevelCategoryIdQueryBuilder);
        }
        if (searchParam.getThirdLevelCategoryId() != null) {
            TermQueryBuilder thirdLevelCategoryIdQueryBuilder = QueryBuilders.termQuery("thirdLevelCategoryId", searchParam.getThirdLevelCategoryId());
            //1.3.4 将关键字查询添加到bool_Query的filter中
            boolQueryBuilder.filter(thirdLevelCategoryIdQueryBuilder);
        }

        //1.4 查询平台属性名和平台属性值
        //1.4.1 先判断前端是否传入了平台属性名和平台属性值
        if (searchParam.getProps() != null && searchParam.getProps().length > 0) {

            //循环遍历前端页面传来的平台属性名和平台属性值
            for (String prop : searchParam.getProps()) {

                //1.4.2 获取前端传来的平台属性名和平台属性值
                String[] platformAttr = prop.split(":");
                Long attrId = Long.valueOf(platformAttr[0]);
                String attrValue = platformAttr[1];
                String attrName = platformAttr[2];

                //1.4.3 构建内层嵌套查询
                BoolQueryBuilder subBoolQueryBuilder = QueryBuilders.boolQuery();
                subBoolQueryBuilder.filter(QueryBuilders.termQuery("attrs.attrId", attrId))
                        .filter(QueryBuilders.termQuery("attrs.attrValue", attrValue));
                //1.4.4 将内层嵌套放入外层中
                // 每一个平台属性的查询都是一个Nested类型的查询
                // 有多少个平台属性的查询，就有多少个Nested查询，每一个Nested查询，都应该放到Bool查询的 filter部分
                NestedQueryBuilder nestedQueryBuilder = QueryBuilders.nestedQuery("attrs", subBoolQueryBuilder, ScoreMode.None);

                //1.4.5 将关键字查询添加到bool_Query的filter中
                boolQueryBuilder.filter(nestedQueryBuilder);
            }
        }

        //1.5 将上述查询条件放入搜索查询请求体对象中
        nativeSearchQueryBuilder.withQuery(boolQueryBuilder);

        //2. 构建分页查询
        //2.1 构建pageRequest对象，需要注意的是: pageRequest.of(page,size) 中的page是从0开始的，我们传入的参数 页面是从1开始的
        PageRequest pageRequest = PageRequest.of(searchParam.getPageNo() - 1, searchParam.getPageSize());
        //2.2 将上述分页条件放入搜索查询请求体对象中
        nativeSearchQueryBuilder.withPageable(pageRequest);

        //3. 构建排序
        //3.1 先判断前端是否传来了排序条件
        if (StringUtils.isNotBlank(searchParam.getOrder())) {
            String[] orderString = searchParam.getOrder().split(":");
            String orderField = orderString[0].equals("1") ? "hotScore" : "price";
            String orderPricinple = orderString[1];
            Sort.Direction direction = orderPricinple.equals("asc") ? Sort.Direction.ASC : Sort.Direction.DESC;

            //3.2 将上述排序条件放入搜索查询请求体对象中
            nativeSearchQueryBuilder.withSort(Sort.by(direction, orderField));

        } else {
            //3.3 假如前端没有传入排序条件，我们添加默认排序规则
            nativeSearchQueryBuilder.withSort(Sort.by(Sort.Direction.ASC, "hotScore"));
        }

        //4. 构建聚合
        //4.1 构建品牌分桶聚合查询（从外至内，terms下手）
        TermsAggregationBuilder tmAggregationBuilder = AggregationBuilders.terms("tmId_bucket").field("tmId")
                .subAggregation(AggregationBuilders.terms("tmName_bucket").field("tmName"))
                .subAggregation(AggregationBuilders.terms("tmUrl_bucket").field("tmLogoUrl"));
        //将品牌的聚合查询条件放入搜索查询请求体对象中
        nativeSearchQueryBuilder.withAggregations(tmAggregationBuilder);

        //4.2 构建平台属性聚合查询（从外至内，nested—>terms）
        NestedAggregationBuilder attrsAggregationBuilder = AggregationBuilders.nested("attrsAgg", "attrs")
                .subAggregation(AggregationBuilders.terms("attrIdAgg_bucket").field("attrs.attrId")
                        .subAggregation(AggregationBuilders.terms("attrNameAgg_bucket").field("attrs.attrName"))
                        .subAggregation(AggregationBuilders.terms("attrValueAgg_bucket").field("attrs.attrValue"))
                );
        //将平台属性的聚合查询条件放入搜索查询请求体对象中
        nativeSearchQueryBuilder.withAggregations(attrsAggregationBuilder);

        //5. 构建高亮查询
        //先判断前端是否传入高亮的关键字
        if (StringUtils.isNotBlank(searchParam.getKeyword())) {
            //构建高亮查询条件
            HighlightBuilder highlightBuilder = new HighlightBuilder().field("title").preTags("<span style='color:red'>").postTags("</span>");

            //将高亮查询条件放入搜索查询请求体对象中
            nativeSearchQueryBuilder.withHighlightBuilder(highlightBuilder);
        }

        //6. 构建选择字段进行查询（查询出商品的显示的基本信息）
        nativeSearchQueryBuilder.withSourceFilter(new FetchSourceFilter(new String[]{"id", "defaultImg", "title", "price"}, null));

    }

    /**
     * 解析搜索的结果
     *
     * @param searchHits
     * @param searchParam
     */
    private SearchResponseDTO parseResponse(SearchHits<Goods> searchHits, SearchParam searchParam) {

        //1. 将解析的结果封装到SearchResponseDTO中
        SearchResponseDTO searchResponseDTO = new SearchResponseDTO();

        //2. 解析聚合的结果
        Aggregations aggregations = (Aggregations) searchHits.getAggregations().aggregations();
        //2.1 获取品牌聚合的map数据转换为List<SearchResponseTmDTO>
        Terms tmIdBucket = aggregations.get("tmId_bucket");
        //2.2 解析品牌聚合为一个list，将品牌聚合bucket转换为SearchResponseTmDTO
        List<SearchResponseTmDTO> trademarkList = tmIdBucket.getBuckets().stream().map(bucket -> {
            //2.2.1 获取品牌的id
            long tmId = bucket.getKeyAsNumber().longValue();
            //获取子聚合的map数据
            Map<String, Aggregation> map = bucket.getAggregations().asMap();
            //2.2.2 获取品牌名称的桶信息
            Terms tmNameBucket = (Terms) map.get("tmName_bucket");
            String tmName = tmNameBucket.getBuckets().get(0).getKeyAsString();
            //2.2.3 获取品牌url的桶信息
            Terms tmUrlBucket = (Terms) map.get("tmUrl_bucket");
            String tmLogoUrl = tmUrlBucket.getBuckets().get(0).getKeyAsString();

            return new SearchResponseTmDTO(tmId, tmName, tmLogoUrl);
        }).collect(Collectors.toList());


        //2.3 获取平台属性聚合的map数据转换为List<SearchResponseAttrDTO>
        Nested attrsAgg = aggregations.get("attrsAgg");
        //从nested对象中获取子聚合的map信息
        Map<String, Aggregation> map = attrsAgg.getAggregations().asMap();
        //获取平台属性id的桶聚合信息
        Terms attrIdAggBucket = (Terms) map.get("attrIdAgg_bucket");
        //通过流将List<bucket>转换为List<SearchResponseAttrDTO>
        List<SearchResponseAttrDTO> attrsList = attrIdAggBucket.getBuckets().stream().map(bucket -> {
            //2.3.1 获取平台属性id
            long attrId = bucket.getKeyAsNumber().longValue();
            //2.3.2 获取平台属性名称的桶聚合信息
            Terms attrNameAggBucket = (Terms) bucket.getAggregations().get("attrNameAgg_bucket");
            String attrName = attrNameAggBucket.getBuckets().get(0).getKeyAsString();
            //2.3.3 获取平台属性值的桶聚合信息
            Terms attrValueAggBucket = (Terms) bucket.getAggregations().get("attrValueAgg_bucket");
            List<String> attrValueList = attrValueAggBucket.getBuckets().stream().map(attrValueBucket -> {
                String attrValue = attrValueBucket.getKeyAsString();
                return attrValue;
            }).collect(Collectors.toList());

            //2.3.4 将上述信息封装到SearchResponseAttrDTO
            return new SearchResponseAttrDTO(attrId, attrValueList, attrName);
        }).collect(Collectors.toList());

        //3. 解析商品信息List<GoodsDTO>,如果有高亮，要解析高亮
        List<SearchHit<Goods>> searchHitsGoods = searchHits.getSearchHits();
        //3.1 通过流将List<SearchHit<Goods>>转换为List<GoodsDTO>
        List<Goods> goodsPOList = searchHitsGoods.stream().map(searchHit -> {
            Goods goods = searchHit.getContent();

            //解析高亮字符串
            List<String> titleList = searchHit.getHighlightField("title");
            if (CollectionUtils.isNotEmpty(titleList)) {
                String title = titleList.get(0);
                //将高亮字符串放入goods中
                goods.setTitle(title);
            }

            return goods;
        }).collect(Collectors.toList());
        //3.2 将List<Goods>转换为List<GoodsDTO>
        List<GoodsDTO> goodsList = goodsConverter.goodsPOs2DTOs(goodsPOList);

        //4. 解析获取总记录数
        long total = searchHits.getTotalHits();

        //5. 解析页面大小
        Integer pageSize = searchParam.getPageSize();

        //6. 解析获取当前页码
        Integer pageNo = searchParam.getPageNo();

        //7. 解析获取总页数
        Long totalPages = (long) Math.ceil(total / pageSize);
        if (totalPages == 0) {
            totalPages = 1L; //因为前端页数从1开始的。
        }

        return new SearchResponseDTO(trademarkList, attrsList, goodsList, total, pageSize, pageNo, totalPages);
    }
}
